"""
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    created by lane.chang on '07/09/2022'
    comment: 封装的最底层函数
"""
import os
import re
import random
import string
import datetime
import threading
import math
import asyncio
from decimal import Decimal
from importlib import import_module
from typing import Any, Optional, Union
from inspect import getmembers
from urllib.parse import urlparse
from enum import Enum
from pydantic import Field
from laner.pydantic import BaseModel

import pypinyin
from starlette_context import context
from fastapi.responses import JSONResponse

from project.core.exception import DEFAULT_ERROR_CODES


# 数字汉字
NUM_CHR = {
    0: '零',
    1: '一',
    2: '二',
    3: '三',
    4: '四',
    5: '五',
    6: '六',
    7: '七',
    8: '八',
    9: '九',
    10: '十'
}

# 全国省/市编号列表
province_codes = {
    '11', '12', '13', '14', '15', '21', '22', '23', '31', '32',
    '33', '34', '35', '36', '37', '41', '42', '43', '44', '45',
    '46', '50', '51', '52', '53', '54', '61', '62', '63', '64',
    '65', '71', '81', '82', '91'
}


def get_request_id() -> str:
    return context.data["X-Request-ID"]


def get_process_time() -> float:
    # return eval(context.data["X-Process-Time"]) * 10**6
    return 1


class ResponseModel(BaseModel):
    """API返回对象"""

    code: int = Field(default=0, description="状态码")
    message: str = Field(default="success", description="消息描述")
    # request_id: str = Field(default_factory=get_request_id, description="")
    data: Optional[Union[dict, list, Any]] = Field(
        default=None, description="数据内容"
    )
    exec_time: float = Field(default=0.0, description="逻辑执行时长(ms)")


def handler_text(text: str, length: int, remain_eof: bool = True) -> str:
    """对超长文本进行缩略处理
    :param text:
    :param length:
    :param remain_eof: 是否保留原结尾符
    :return:
    """
    if not text:
        return text
    if len(text) <= length:
        return text

    if remain_eof:
        text = text.replace(text[length:-1], "...")
    else:
        text = text.replace(text[length:], "...")
    return text


def serializer(
    data: Optional[Union[tuple, list, set,
                         dict]]) -> Optional[Union[dict, list]]:
    """递归序列化: 有几种基本类型, 比如时间,日期, decimal 需要序列化
    :param data:
    :return:
    """
    if data is None:
        return None

    # 如果是list或者tuple set
    if isinstance(data, (tuple, list, set)):
        dicts = []
        for each in data:
            dicts.append(serializer(each))
        return dicts
    # 不是dict 就直接返回
    if not isinstance(data, dict):
        return data
    # 将字典中的各种不能序列号的类型转化成字符串
    result_data = {}
    for key, value in data.items():
        if isinstance(value, (tuple, list, dict)):
            # 如果值是list, tuple, dict, 那么需要遍历下去
            result_data[key] = serializer(value)
        elif isinstance(value, datetime.datetime):
            result_data[key] = value.strftime("%Y-%m-%d %H:%M:%S")
            result_data[key + '_' + 'ts'] = int(value.timestamp())
        elif isinstance(value, datetime.date):
            result_data[key] = value.strftime("%Y-%m-%d")
        elif isinstance(value, datetime.time):
            result_data[key] = value.strftime("%H:%M:%S")
        elif isinstance(value, Decimal):
            result_data[key] = str(value)
        elif isinstance(value, Enum):
            result_data[key] = value.value
        else:
            result_data[key] = value

    return result_data


def json_response(
    code: int = 0,
    data: Optional[Union[tuple, list, set, dict, str, bool]] = None,
    message: Optional[str] = None,
) -> JSONResponse:
    """构造api的json返回值
    :param code:
    :param data:
    :param message:
    :return:
    """
    response_model = ResponseModel()

    response_model.code = code

    msg_head = DEFAULT_ERROR_CODES.get(code)
    if message:
        response_model.message = message
    elif msg_head:
        response_model.message = msg_head

    if data is not None:
        response_model.data = serializer(data)

    return JSONResponse(content=response_model.dict())


def get_modules(package_name: str) -> list[str]:
    """获取包名下所有非__init__的模块名
    :param: package_name: 指定包名(含导入路径名)
    """
    package = import_module(package_name)
    path_name = package.__path__[0]
    files = os.listdir(path_name)
    modules = []
    for file in files:
        if file.startswith("__"):
            continue
        if os.path.isfile(f"{path_name}/{file}"):
            if not file.endswith(".py"):
                continue
            name, _ = os.path.splitext(file)
            if not package_name:
                modules.append(name)
            else:
                modules.append(f"{package_name}.{name}")
        else:
            modules.extend(get_modules(package_name=f"{package_name}.{file}"))

    return modules


def get_models(package_name: str, predicate=None) -> list[Any]:
    """获取包名下所有指定类
    :param: package_name: 指定包名(含导入路径名)
    """
    modules = get_modules(package_name)
    ret_models = []
    for module in modules:
        _module = import_module(module)
        ret_models.extend(getmembers(_module, predicate))

    return [v for _, v in ret_models]


def handler_text(text: str, length: int, remain_eof: bool = True) -> str:
    """对超长文本进行缩略处理
    :param text:
    :param length:
    :param remain_eof: 是否保留原结尾符
    :return:
    """
    if not text:
        return text
    if len(text) <= length:
        return text

    if remain_eof:
        text = text.replace(text[length:-1], "...")
    else:
        text = text.replace(text[length:], "...")
    return text


def generate_path(file_path: str = ""):
    """检查并生成生成文件路径
    :param file_path:
    :return:
    """
    if not os.path.exists(file_path):
        os.makedirs(file_path)

    return file_path


def generate_random_letter():
    """生成随机字母
    :return:
    """
    return random.choice(string.ascii_lowercase)


def get_login_key():
    """ 根据人员类型, 返回不同的login_key
    :return:
    """
    return 'login_token_user'


def set_login_info(rds, user_id, token):
    """
    :desc: 设置登录用户的信息(含刷新)
    :return:
    """
    rds.hset(name=get_login_key(), key=user_id, value=token)


def delete_login_info(rds, user_id):
    """
    :desc: 删除用户登陆信息
    :return:
    """
    rds.hdel(get_login_key(), user_id)


def convert_py(text: Union[Any], separator: str = '', ret_form: str = 'str', style=pypinyin.NORMAL) -> Union[str, list]:
    """ 转拼音
    :param: text: 文本
    :param: separator: 分隔符
    :param: ret_form: 返回形式
    :param: style: 拼音类型 normal(无声调)/tone(有声调)
    :return:
    """
    try:
        if type(text) != str:
            return text
        text_py = pypinyin.pinyin(text, style=style)
        if ret_form == 'str':
            return separator.join([v[0] for v in text_py])

        if ret_form == 'list':
            return [v[0] for v in text_py]
    except Exception as ex:
        print(ex)
        return text


def pop_elements(_list, count=1, is_random=True):
    """
    抛出指定数量的元素

    Args:
        _list(list): 列表
        count(int): 抛出元素的位数量
        is_random(bool): 是否随机

    Raises:
        IndexError

    Returns:
        列表抛出的结果
    """
    # 判断是否为空
    if not _list:
        return
    if len(_list) < count:
        count = len(_list)
    # 是否随机
    if is_random:
        return [_list.pop(random.choice(range(0, len(_list)))) for _ in range(count)]

    return [_list.pop(-1) for _ in range(count)]


def start_new_thread() -> asyncio.AbstractEventLoop:
    """ 创建并启动新线程 - 实现异步
    :return:
    """
    new_loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()

    def loop_in_thread(_loop):  # 一个将被丢进线程的函数
        asyncio.set_event_loop(_loop)  # 调用loop需要使用set_event_loop方法指定loop
        _loop.run_forever()  # run_forever() 会永远阻塞当前线程，直到有人停止了该loop为止。

    # 开启客户端接收
    t = threading.Thread(target=loop_in_thread, args=(new_loop,))
    t.start()

    return new_loop


def format_seconds(value: int) -> str:
    """ 格式化秒
    :return:
    """
    ret_str = ''
    # 时
    hour = value // 3600
    if hour > 0:
        ret_str += f'{hour}时'

    # 分
    minutes = (value - hour * 3600) // 60
    if minutes > 0:
        ret_str += f'{minutes}分'

    # 秒
    seconds = value - hour * 3600 - minutes * 60
    if seconds > 0:
        ret_str += f'{seconds}秒'

    if not ret_str:
        ret_str = '0秒'

    return ret_str


def is_url_urllib(str_value):
    """
    判断字符串是否为URL的urllib库方法
    """
    try:
        result = urlparse(str_value)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False


def num_convert_chr(value: str) -> str:
    """ 数值转汉字
    :return:
    """
    ret = ''
    for v in value:
        if v.isdigit():
            ret += NUM_CHR.get(int(v))
        else:
            ret += v

    return ret


def is_subsequence(content: Union[list, str], text: Union[list, str]):
    """ 是否按顺序包含
    :param content: 小文本
    :param text: 大文本
    :return:
    """
    text = iter(text)
    return all(v in text for v in content)


def match_index(text: Union[str, list, tuple], content: Union[str, int]) -> int:
    """ 返回匹配内容在文本中的位置
    :param text: 大文本
    :param content:
    :return: 
    """
    try:
        index = text.index(content)
    except ValueError as ex:
        pass
        index = -1

    return index


def cal_age_with_birth_ym(birth_ym: str):
    """
    :param birth_ym:
    :return:
    """
    start_date = None
    if len(birth_ym) == 7:
        start_date = datetime.datetime.strptime(birth_ym, '%Y-%m')
    elif len(birth_ym) == 6:
        start_date = datetime.datetime.strptime(birth_ym, '%Y%m')

    age = 0
    if start_date:
        end_date = datetime.datetime.now()
        months = (end_date.year - start_date.year) * 12 + end_date.month - start_date.month
        age = math.floor(months / 12) if months > 0 else 0

    return age


def get_intersection_text(text_compared: str, text: str, exact=True, style=pypinyin.NORMAL, mode='complete') -> dict:
    """ 返回文本交集匹配的部分(从字符串的开始部分匹配)
    :param text_compared: 要对照的文本
    :param text:
    :param exact: 是否精确匹配(主要包括拼音匹配逻辑)
    :param style:
    :param mode:
    :return:
    """
    match_text = ''
    text_updated = text_compared
    match_index = -1
    ret_data = {
        'match_text': match_text,
        'text_updated': text_updated,
        'match_index': match_index
    }
    if exact:
        count = 0
        while True:
            if count == 0:
                text_tmp = text
            else:
                text_tmp = text[0: count]

            if text_tmp in text_compared:
                match_text = text_tmp
                match_index = text_compared.index(text_tmp)
                text_updated = text_compared.replace(match_text, '')
                break

            if int(math.fabs(count)) >= len(text) - 1:
                break

            count -= 1
    else:
        _texts_compared = convert_py(num_convert_chr(text_compared), ret_form='list', style=style)
        texts = convert_py(num_convert_chr(text), ret_form='list', style=style)

        intersection_list, start_index = get_intersection_list(texts_compared=_texts_compared, texts=texts)

        for index in range(len(intersection_list)):
            match_text += text[index]
            match_index = start_index

        text_updated = text_compared.replace(match_text, '')

    if mode == 'complete' and match_text != text:
        return ret_data

    ret_data['match_text'] = match_text
    ret_data['text_updated'] = text_updated
    ret_data['match_index'] = match_index

    return ret_data


def get_intersection_list(texts_compared: list[Any], texts: list[Any]) -> (list[Any], int):
    """ 返回列表交集匹配部分
    :param texts_compared: 要对照的列表
    :param texts:
    :return:
    """
    # if len(texts) > len(texts_compared):
    #     return []

    match_indexes = []
    texts_compared_tmp = texts_compared
    index_tmp = 0
    while True:
        text = texts[0]

        index = match_index(texts_compared_tmp, text)
        if index == -1:
            break

        match_indexes.append(index_tmp + index)
        index_tmp += index + 1
        texts_compared_tmp = texts_compared_tmp[index + 1:]

    match_texts = []
    start_index = -1
    num = 0
    for index in match_indexes:
        text_compared = texts_compared[index]
        text = texts[0]
        count = 1
        while True:
            if index + count > len(texts_compared) - 1:
                count -= 1
                break
            text_compared += texts_compared[index + count]

            if count > len(texts) - 1:
                count -= 1
                break
            text += texts[count]

            if text_compared != text:
                count -= 1
                break
            count += 1

        if len(match_texts) <= count + 1:
            match_texts = texts[0: count + 1]
            start_index = index
            num = count + 1

    if start_index > -1:
        for i in range(start_index + num - 1, start_index - 1, -1):
            texts_compared.pop(i)

    return match_texts, start_index


def get_match_indexes(content: Union[list, str], text: Union[list, str]) -> Optional[list[int]]:
    """ 返回文本公共部分index列表
    :param content: 小文本
    :param text: 大文本
    :return:
    """
    if not is_subsequence(content, text):
        return

    indexes = []
    for index, v in enumerate(text):
        if v in content:
            indexes.append(index)

        if match_index(content, v) >= len(content) - 1:
            break

    return indexes


# 判断字符是否为汉字
def is_chinese(word):
    """ 判断是否为汉字
    :param word:
    :return:
    """
    pattern = re.compile(r'[^\u4e00-\u9fa5]')
    if pattern.search(word):
        return False
    else:
        return True


def validate_id_number(id_number: str) -> bool:
    """ 验证身份证号码是否合法
    :param id_number:
    :return:
    """
    if not id_number:
        return False
    # 验证身份证号码格式是否正确
    pattern = r'^\d{17}[\dXx]$'
    if not re.match(pattern, id_number):
        return False

    # 验证身份证号码中的地区码是否正确
    province_code = id_number[:2]
    if province_code not in province_codes:
        return False

    # 验证身份证号码中的出生日期是否正确
    birthday = id_number[6:14]
    try:
        datetime.datetime.strptime(birthday, '%Y%m%d')
    except ValueError:
        return False

    # 验证身份证号码中的校验码是否正确
    factors = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
    check_sum = sum(int(id_number[i]) * factors[i] for i in range(17))
    check_code = '10X98765432'[check_sum % 11]
    if check_code != id_number[-1].upper():
        return False

    # 身份证号码验证通过
    return True


def if_none(value: Any, ret_value: Union[str, int] = '') -> Union[str, int]:
    """ 如果为None, 则返回指定的字符串
    :param value:
    :param ret_value: 指定的
    :return:
    """
    return ret_value if value is None else value


def validate_phone(phone: str) -> bool:
    """ 验证身份证号码是否合法
    :param phone:
    :return:
    """
    pattern = re.compile(r'^(13[0-9]|14[0|5|6|7|9]|15[0|1|2|3|5|6|7|8|9]|'
                         r'16[2|5|6|7]|17[0|1|2|3|5|6|7|8]|18[0-9]|'
                         r'19[1|3|5|6|7|8|9])\d{8}$')

    return bool(re.match(pattern, phone))


def camel_to_snake(value):
    """ 字符串驼峰规则转长蛇规则
    :param value:
    :return:
    """
    result = ''
    for i in range(len(value)):
        if value[i].isupper():
            result += '_' + value[i].lower()
        else:
            result += value[i]

    return result.lstrip('_')


def split_text(text, length=30):
    """ 根据长度截取字符串
    :param text:
    :param length:
    :return:
    """
    texts = []
    for i in range(0, len(text), length):
        texts.append(text[i:i + length])

    return texts


def extract_phone_numbers(text):
    """ 从文本中筛选出手机号信息
    :param text:
    :return:
    """
    pattern = r'1[3456789]\d{9}'

    phone_numbers = re.findall(pattern, text)
    return phone_numbers


def to_dict(exclude_unset=False, **kwargs):
    """
    :param kwargs:
    :param exclude_unset:
    :return:
    """
    if not exclude_unset:
        return kwargs
    else:
        ret_kwargs = dict()
        for k, v in kwargs.items():
            if v is None:
                continue
            ret_kwargs[k] = v

        return ret_kwargs


def extract_content_within_braces(text):
    """ 截取{}的内容
    :param text:
    :return:
    """
    start = text.find('{')
    end = text.find('}')
    if start != -1 and end != -1:
        return text[start+1:end]


def extract_outermost_braces(text):
    # 匹配最外层的大括号内容
    pattern = r'({([^{}]|[^{}]|[^}]*}|[^}]*))|(\[([^\[]]|[^\[]]|[^\]]*)|[^\]]*)'
    match = re.search(pattern, text)
    if match:
        return match.group()
    return None


def extract_content_within_tags(text, start_tag='', end_tag=''):
    """
    :param text:
    :param start_tag:
    :param end_tag:
    :return:
    """
    start = text.find(start_tag)
    end = text.find(end_tag)

    if start == -1 or end == -1:
        return text

    start = start + len(start_tag) - 1

    return text[start+1: end]


def match_text(text, tag, replace: tuple=()):
    """ 匹配标签的内容
    :param text:
    :param tag:
    :param replace: 设置替换
    :return:
    """
    # 正则表达式模式，匹配<tag>和</tag>之间的内容
    pattern = rf'<{tag}>(.*?)</{tag}>'

    # 使用re.search查找匹配的内容
    match = re.search(pattern, text)

    if match:
        # 打印匹配的staff_id内容
        if not replace:
            return match.group(1)
        else:
            return match.group(1).replace(*replace)
    else:
        return ''


def match_texts(text, *args, replace: tuple = ()):
    """
    :param text:
    :param args:
    :param replace: 设置替换
    :return:
    """
    result = {}
    for arg in args:
        result[arg] = match_text(text, arg, replace)

    return result


def find_value(target, data, not_found=''):
    """ 查找字典指定的key值(递归)
    :param target:
    :param data:
    :param not_found:
    :return:
    """
    queue = [data]
    while len(queue) > 0:
        data = queue.pop()
        for key, value in data.items():
            if key == target:
                return value
            elif type(value) == dict:
                queue.append(value)
    return not_found


class Core(BaseModel):
    """
    """
    func: Any = Field(..., title='异步函数')
    name: str = Field(..., title='函数名')

async def async_run(*args):
    """ 异步执行任务
    :param args:
        core:
        name:
    :return:
    """
    ret_data = dict()
    tasks = list()
    for core in args:
        tasks.append(asyncio.create_task(core.func, name=core.name))

    await asyncio.gather(*tasks)

    for task in tasks:
        ret_data[task.get_name()] = task.result()

    return ret_data
